#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import json
import locale
import os.path
import platform
import signal
import threading
import time
from pexpect import popen_spawn
from common.const import ExecuteResultEnum
from common.common import read_result_file
from common.file_common import delete_file, create_dir_recursive
from oracle import logger
from oracle.common.common import check_cdb_is_open, exec_sql_cmd, log_func_time_cost, \
    write_tmp_file, parse_html_result, login_oracle_database, \
    get_database_info, \
    get_compatible_info, \
    chown_file_path_owner, \
    check_table_exist, checkout_user_and_execute_cmd, escape_special_chars, check_del_dir_or_file
from oracle.common.constants import PexpectResult, ScriptExitCode, PluginPathConstant, ArchiveLogMode, OpenMode, \
    ErrorCode, OracleDataBaseType
from oracle.common.linux_common import execute_linux_rman_cmd
from oracle.common.sql_common import create_new_temp_table_space_sql
from oracle.common.user_env_common import get_oracle_user_env_by_linux
from oracle.common.windows_common import redirect_backup_path
from common.common_models import ActionResult
from oracle.services.restore.restore_common import login_oracle_database_restore, exec_oracle_sql_cmd, shutdown_abort, \
    get_pdb_name_set, get_violation_message_set
from common.parse_parafile import get_env_variable


class TableOracleRestore:
    def __init__(self, job_id, pid, tables, sub_job_id, is_overwrite, assist_ins_sid=None):
        self.login_params = {}
        self.assist_ins_sid = assist_ins_sid
        self.restore_params = {}
        self.ora_version = ''
        self.sql_params = {}
        self.rman_params = {}
        self.db_is_cluster = 0
        self.additional = ''
        self.oracle_is_cdb = ''
        self.auth_type = ''
        self.spfile_location_new = ''
        self.spfile_location = ''
        self.omf_enable = ''
        self.ora_preversion_extend = ''
        self.ora_preversion = ''
        self.database_role = ''
        self.oracle_base = ''
        self.ora_db_group = ''
        self.task_type = 0
        self.resetlogs_id = ''
        self.pfile_name = ''
        self.dbpw_file = ''
        self.upgrade = ''
        self.incar_num = ''
        self.backup_all_version = ''
        self.backup_pre_version = ''
        self.export_oracle_env = ''
        self.rman_enc_section = ''
        self.is_enc_bk = 0
        self.pit = ''
        self.copy_restore = 0
        self.main_backup_path = ''
        self.backup_path_list = ''
        self._pid = pid
        self._sub_job_id = sub_job_id
        self._job_id = job_id
        self.db_name = ''
        self.db_instance = ''
        self.db_user = ''
        self.db_user_pwd = ''
        self.oracle_home = ''
        self.asm_user = ''
        self.asm_user_pwd = ''
        self.channels = 0
        self.pit_time = ''
        self.pit_scn = ''
        self.recover_target = ''
        self.recover_path = ''
        self.recover_order = ''
        self.storage_type = ''
        self.asm_sid_name = ''
        self.db_type = ''
        self.enc_algo = ''
        self.enc_key = ''
        self.pfile_pid = ''
        self.restore_by = ''
        self.recover_num = ''
        self.ora_db_user = ''
        self.ora_grid_user = ''
        self.backup = ''
        self.meta_data_path = ''
        self.archive = ''
        self.job_id_list = ''
        self.bct_status = ''
        self.backup_time = ''
        self.tables = json.loads(tables)
        self.cache_repo_path = ''
        self.is_cdb_aim = None
        self.table_dump_file = {}
        self.is_overwrite = is_overwrite if is_overwrite.lower() == 'true' else False
        self.service_ip = ''
        self.port = ''
        self.target_sub_type = ''
        self.target_db_version = ''

    @log_func_time_cost
    def table_restore(self):
        logger.info(f"start expdp impdp, pid: {self._pid}.")
        self.define_params()

        # 从辅助实例导出表
        ret = self.export_tables()
        if ret != ScriptExitCode.SUCCESS:
            logger.error(f'export_tables failed, ret: {ret}')
            return ret

        # 从辅助实例导入表
        ret = self.import_tables()
        if ret != ScriptExitCode.SUCCESS:
            return ret

        return ScriptExitCode.SUCCESS

    @log_func_time_cost
    def export_tables(self):
        """
        表导出逻辑：
            非CDB库：
                1、在cache仓创建导出临时目录
                2、执行创建directory的sql并赋权
                3、执行expdp导出表
            CDB库：
                1、在cache仓创建导出临时目录
                2、解析入参，给每个pdb执行创建directory的sql并赋权
                3、执行expdp导出表

        @return: ScriptExitCode状态
        """
        logger.info(f"start expdp, pid: {self._pid}.")
        # 辅助实例sid
        assist_ins_sid = self.assist_ins_sid
        # 数据泵导出目录名称
        assist_ins_expdp_dir_name = f"{assist_ins_sid}_expdp_dir"
        assist_ins_expdp_dir_path = self.get_assist_ins_expdp_dir_path()
        if self._is_cdb_database_copy():
            # 副本为cdb场景
            pdb_name_list = self.get_pdb_name_list()
            if not pdb_name_list:
                logger.error(f'resolve json_tables to pdb_list empty, json_tables: {self.tables}')
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
            # 打开辅助实例下的pdb
            if not self._open_all_pdbs():
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
            # 为辅助实例创建临时表空间
            for pdb_name in pdb_name_list:
                self.create_new_temp_table_space(pdb_name)
            # 在辅助实例各个PDB下执行sql创建导出目录
            db_login_params = {
                'auth_pwd': f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
                'auth_key': f"job_copies_0_protectObject_auth_authKey_{self._pid}"
            }
            pdb_list = self._create_pdb_directories(
                db_login_params, self.assist_ins_sid, assist_ins_expdp_dir_name, assist_ins_expdp_dir_path,
                pdb_name_list
            )
            if not pdb_list:
                logger.error(f'resolve json_tables to pdb_list empty, json_tables: {self.tables}')
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
            target_pdb_names = self._get_target_pdb_names()
            ret = self._export_cdb_tables(assist_ins_expdp_dir_name, assist_ins_expdp_dir_path, pdb_list)
            if ret != ScriptExitCode.SUCCESS:
                logger.error(f'_export_cdb_tables failed {ret}')
                self.delete_directories(assist_ins_expdp_dir_name, target_pdb_names)
                return ret
            # 导出成功后关闭辅助实例
            self.shutdown_assist_ins()
            return ScriptExitCode.SUCCESS
        else:
            # 副本为非cdb场景
            # 在辅助实例执行sql创建导出目录
            logger.info(f"start expdp for non cdb, pid: {self._pid}.")
            db_login_params = {
                'auth_pwd': f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
                'auth_key': f"job_copies_0_protectObject_auth_authKey_{self._pid}"
            }
            ret = self._create_directories(
                db_login_params, assist_ins_sid, assist_ins_expdp_dir_name, assist_ins_expdp_dir_path
            )
            if ret != ScriptExitCode.SUCCESS:
                return ret
            # 为辅助实例创建临时表空间
            self.create_new_temp_table_space(None)
            # 执行数据泵expdp导出数据到cache仓中
            return self._export_non_cdb_tables(assist_ins_sid, assist_ins_expdp_dir_name)

    def delete_directories(self, assist_ins_expdp_dir_name, target_pdb_names):
        logger.info(f"start delete directories, instance: {self.db_instance}, "
                    f"dir_name:{assist_ins_expdp_dir_name}")
        db_login_params = {
            'auth_pwd': f"job_targetObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_targetObject_auth_authKey_{self._pid}"
        }
        result = self._delete_directories(db_login_params, self.db_instance,
                                          assist_ins_expdp_dir_name, target_pdb_names)
        if result != ScriptExitCode.SUCCESS:
            logger.warn(f'_delete_directories failed')
        return result

    def get_pdb_name_list(self):
        pdb_name_list = set()
        for json_table in self.tables:
            pdb_name = json_table.get('pdb_name', '')
            if pdb_name:
                pdb_name_list.add(pdb_name)
        return pdb_name_list

    def get_assist_ins_expdp_dir_path(self):
        # 数据泵导出目录路径
        if platform.system().lower() == "windows":
            assist_ins_expdp_dir_path = os.path.join(self.oracle_home, f"table_dump_{self._job_id}")
            create_dir_recursive(assist_ins_expdp_dir_path)
        else:
            assist_ins_expdp_dir_path = f"{self.cache_repo_path}/table_dump_{self._job_id}"
            create_dir_recursive(assist_ins_expdp_dir_path)
            chown_file_path_owner(assist_ins_expdp_dir_path, self.ora_db_user)
        return assist_ins_expdp_dir_path

    def shutdown_assist_ins(self):
        db_login_params = {
            'pid': self._pid,
            "instance_name": self.assist_ins_sid,
            "db_install_user": self.ora_db_user,
            'auth_pwd': f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_copies_0_protectObject_auth_authKey_{self._pid}"
        }
        shutdown_abort(db_login_params)

    def create_new_temp_table_space(self, pdb_name):
        logger.info(f"create_new_temp_table_space, pdb_name: {pdb_name}")
        login_params = {
            'instance_name': self.assist_ins_sid,
            "db_install_user": self.ora_db_user,
            "auth_pwd": f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
            "auth_key": f"job_copies_0_protectObject_auth_authKey_{self._pid}"
        }
        alter_pdb_cmd = ''
        if pdb_name:
            alter_pdb_cmd = f'alter session set container="{pdb_name}";\n'
        create_new_temp_table_space_sql(self._pid, login_params,
                                        f"{self.cache_repo_path}/new_temp_{self.assist_ins_sid}_{pdb_name}.dbf",
                                        pdb_name, alter_pdb_cmd)

    @log_func_time_cost
    def import_tables(self):
        """
        导入表到目标实例

        @return: ScriptExitCode状态
        """
        logger.info(f"start impdp, pid: {self._pid}.")
        # 数据泵导出目录名称
        expdp_dir_name = f"{self.assist_ins_sid}_expdp_dir"
        # 数据泵导出目录路径
        if platform.system().lower() == "windows":
            expdp_dir_path = os.path.join(self.oracle_home, f'table_dump_{self._job_id}')
        else:
            expdp_dir_path = f"{self.cache_repo_path}/table_dump_{self._job_id}"
        target_pdb_names = set()
        try:
            ret, target_pdb_names = self._exec_import_tables(expdp_dir_name, expdp_dir_path)
            if ret != ScriptExitCode.SUCCESS:
                logger.error(f'_exec_import_tables failed')
                return ret
        finally:
            logger.info(f"start delete directories, instance: {self.db_instance}, dir_name:{expdp_dir_name}")
            db_login_params = {
                'auth_pwd': f"job_targetObject_auth_authPwd_{self._pid}",
                'auth_key': f"job_targetObject_auth_authKey_{self._pid}"
            }
            ret = self._delete_directories(db_login_params, self.db_instance, expdp_dir_name, target_pdb_names)
            if ret != ScriptExitCode.SUCCESS:
                logger.warn(f'_delete_directories failed')
        return ScriptExitCode.SUCCESS

    def get_restore_vars(self, params):
        self.db_name = params['AppName']
        self.db_instance = params['InstanceName']
        self.db_user = params['UserName']
        self.db_user_pwd = params['Password']
        self.oracle_home = params['OracleHome']
        self.oracle_base = params['OracleBase']
        self.asm_user = params['ASMUserName']
        self.asm_user_pwd = params['ASMPassword']
        self.channels = params['Channel']
        self.pit_time = params['pitTime']
        self.pit_scn = params['pitScn']
        self.recover_target = params['recoverTarget']
        self.recover_path = params['recoverPath']
        self.recover_order = params['recoverOrder']
        self.storage_type = params['storType']
        self.asm_sid_name = params['ASMInstanceName']
        self.db_type = params['dbType']
        self.enc_algo = params['EncAlgo']
        self.enc_key = params['EncKey']
        self.pfile_pid = params['pfilePID']
        self.restore_by = params['RestoreBy']
        self.recover_num = params['recoverNum']
        self.ora_db_user = params['OracleInstallUser']
        self.ora_db_group = params['OracleInstallGroup']
        self.ora_grid_user = params['GridInstallUser']
        self.backup = params['DataPath']
        self.meta_data_path = params['MetaDataPath']
        self.archive = params['LogPath']
        self.job_id_list = params['JobIdList']
        self.bct_status = params['bctStatus']
        self.backup_time = params['backupTime']
        self.cache_repo_path = params['cacheRepoPath']
        self.target_db_version = params['targetDbVersion']
        self.service_ip = params['service_ip']
        self.port = params['port']
        self.target_sub_type = params['target_sub_type']
        self.rman_params = self.get_rman_params()
        self.login_params = {
            'pid': self._pid,
            "instance_name": self.db_instance,
            "db_install_user": self.ora_db_user,
            "auth_pwd": f"job_targetObject_auth_authPwd_{self._pid}",
            "auth_key": f"job_targetObject_auth_authKey_{self._pid}"
        }

    def get_rman_params(self):
        return {
            'pid': self._pid,
            "is_enc_bk": 0,
            "instance_name": self.db_instance,
            "db_user": self.db_user,
            "db_password": self.db_user_pwd,
            "rman_enc_section": "",
            "db_install_user": self.ora_db_user,
            "db_install_group": self.ora_db_group
        }

    def define_params(self):
        logger.info(f"table oracle restore start, pid:{self._pid}.")
        self.backup_path_list = redirect_backup_path(self.backup)
        self.main_backup_path = self.backup_path_list.split(";")[0]
        self.additional = f"{self.main_backup_path}/additional"
        backup_temp = f"{self.main_backup_path}/tmp"
        logger.info(f'BACKUPTEMP={backup_temp};ADDITIONAL={self.additional};'
                    f'MainBackupPath={self.main_backup_path}, pid:{self._pid}.')
        if self.channels:
            self.channels = int(self.channels)
        if not self.channels or self.channels == 0:
            self.channels = 4
            logger.warning(F"Setting channels number to {self.channels} by default, pid:{self._pid}.")
        if self.channels > 254:
            logger.error(f"Channels param is invalid, Channels={self.channels}, pid:{self._pid}.")
            return 1
        return ScriptExitCode.SUCCESS

    def table_restore_prerequisite(self):
        base_info = get_database_info(self.login_params, self._job_id)
        if not base_info:
            logger.error(f"Get database base info failed, pid:{self._job_id}")
            return ActionResult(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                bodyErr=ErrorCode.ERROR_INSTANCE_NOSTART.value)
        [log_mode, db_id, uniq_name, open_mode, incarnation_number, self.resetlogs_id] = base_info
        # 归档模式
        if log_mode != ArchiveLogMode.ARCHIVELOG.value:
            logger.error(f"table restore_prerequisite, log mode not archivelog, pid: {self._job_id}.")
            return ActionResult(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                bodyErr=ErrorCode.ERROR_ORACLE_NOARCHIVE_MODE.value)
        # 读写模式
        if open_mode != OpenMode.READWRITE.value:
            logger.error(f"table restore_prerequisite, open mode not readwrite, pid: {self._job_id}.")
            return ActionResult(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                bodyErr=ErrorCode.ERROR_ORACLE_NO_READWRITE_MODE.value)
        # compatible配置需要11.1.0以上 源端，目标端
        compatible = get_compatible_info(self._pid, self.login_params)
        compatible_versions = compatible.split(".")
        if (int(compatible_versions[0]) < 12
                or (int(compatible_versions[0]) == 11 and int(compatible_versions[1]) < 1)):
            logger.error(f"compatible_versions is lower 11.1.0")
            return ActionResult(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                bodyErr=ErrorCode.ERROR_ORACLE_COMPATIBLE_VERSION_LOWER.value)
        # 不开覆盖时，恢复的表不能已存在
        if not self.is_overwrite:
            json_tables = self.tables
            for table in json_tables:
                table_name = table.get('target_table_name', '')
                user_name = table.get('target_user_name', '')
                pdb_name = table.get('target_pdb_name', '')
                table_exist = check_table_exist(self._pid, self.login_params, user_name, table_name,
                                                pdb_name)
                if table_exist:
                    tables = list()
                    tables.append(table_name)
                    logger.error(f"table {table_name} exits")
                    return ActionResult(code=ExecuteResultEnum.INTERNAL_ERROR.value,
                                        bodyErr=ErrorCode.ERROR_EXIST_SAME_TABLE_NAME.value, bodyErrParams=tables)
        return ActionResult(code=ExecuteResultEnum.SUCCESS.value)

    def is_version_later_than_19c(self):
        logger.info(f'is_version_later_than_19c {self.target_db_version}')
        item = str(self.target_db_version).split('.')[0]
        return int(item) >= 19

    def _get_target_pdb_names(self):
        tables = self.tables
        target_pdb_names = set()
        for table in tables:
            target_pdb_name = table.get('target_pdb_name', '')
            if not target_pdb_name:
                continue
            target_pdb_names.add(target_pdb_name)
        return target_pdb_names

    def _exec_dump_pdb_tables(self, assist_ins_sid, pdb):
        # 执行pdb导出，失败重试20次
        exec_cnt = 0
        return_code = 0
        while exec_cnt < 20:
            if self.is_version_later_than_19c():
                return_code = self._dump_pdb_tables_19c(assist_ins_sid, pdb)
            else:
                return_code = self._dump_pdb_tables(assist_ins_sid, pdb)
            if return_code == ScriptExitCode.SUCCESS:
                logger.info(f"exec_dump_pdb_tables SUCCESS {exec_cnt}")
                return return_code
            exec_cnt += 1
            time.sleep(5)
        logger.error(f"exec_dump_pdb_tables error")
        return return_code

    def _exec_imp_pdb_table(self, table):
        # 执行pdb导入，失败重试5次
        exec_cnt = 0
        return_code = 0
        while exec_cnt < 5:
            if self.is_version_later_than_19c():
                return_code = self._imp_table_to_pdb_19c(table)
            else:
                return_code = self._imp_table_to_pdb(table)
            if return_code == ScriptExitCode.SUCCESS:
                logger.info(f"exec_imp_pdb_table SUCCESS {exec_cnt}")
                return return_code
            exec_cnt += 1
            time.sleep(5)
        logger.error(f"exec_imp_pdb_table error")
        return return_code

    def _table_exists(self, table):
        target_table_name = table.get('target_table_name', '')
        target_user_name = table.get('target_user_name', '')
        target_pdb_name = table.get('target_pdb_name', '')
        return check_table_exist(self._pid, self.login_params, target_user_name, target_table_name, target_pdb_name)

    def _drop_table(self, table):
        target_table_name = table.get('target_table_name', '')
        target_user_name = table.get('target_user_name', '')
        target_pdb_name = table.get('target_pdb_name', '')
        table_full_name = f"{target_user_name}.\"{target_table_name}\""
        logger.info(f"start drop table: {table_full_name}")
        db_login_params = {
            'auth_pwd': f"job_targetObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_targetObject_auth_authKey_{self._pid}"
        }
        result, child = login_oracle_database_restore(self._pid, self.db_instance, db_login_params, self.ora_db_user)
        sql = f"drop table {table_full_name} CASCADE CONSTRAINTS;"
        expect_result = PexpectResult.TABLE_DROPPED
        if target_pdb_name:
            sql = (f"alter session set container={target_pdb_name};\n"
                   f"drop table {table_full_name};")
            expect_result = PexpectResult.PDB_TABLE_DROPPED
        return_code, std_out = exec_sql_cmd(child, sql, expect_result)
        if not return_code:
            logger.error(f'_drop_table error, return_code: {return_code}. sql: {sql}, stdout:{std_out}')
            return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        return ScriptExitCode.SUCCESS

    def _dump_pdb_tables(self, assist_ins_sid, pdb):
        """
        dump pdb数据库的表

        @param assist_ins_sid: 实例id
        @return: ScriptExitCode状态
        """
        pdb_name = pdb.get('pdb_name')
        pdb_user = pdb.get('pdb_user')
        dir_name = pdb.get('dir_name')

        auth_scope = f"{pdb_user}/{pdb_name}@{pdb_name}"
        # 合并表结构 user_and_table_str ="u1.t1, u2,t2"
        user_and_table_str_list = []
        dumpfile_name = f"{pdb_name}_{self._pid}.dmp"
        for json_table in self.tables:
            if json_table.get('pdb_name') == pdb_name:
                user_and_table = f"{json_table.get('user_name')}.{json_table.get('table_name')}"
                table_name = escape_special_chars(json_table.get('table_name'))
                self.table_dump_file[user_and_table] = dumpfile_name
                if platform.system().lower() == "windows":
                    user_and_table_str_list.append(f"{json_table.get('user_name')}.\"{json_table.get('table_name')}\"")
                else:
                    user_and_table_str_list.append(f"{json_table.get('user_name')}.\\\"{table_name}\\\"")
        user_and_table_str = ",".join(user_and_table_str_list)

        if self.target_sub_type == OracleDataBaseType.ORACLE_CLUSTER:
            userid = f"{pdb_user.upper()}/{pdb_name}@{self.service_ip}:{self.port}/{pdb_name}"
        else:
            userid = f"{pdb_user.upper()}/{pdb_name}@{pdb_name}"
        if platform.system().lower() == "windows":
            expdp_params = {
                'auth_scope': f'USERID={userid}',
                'directory_scope': f'DIRECTORY={dir_name}',
                'dumpfile_scope': f'DUMPFILE={dumpfile_name}',
                'tables_scope': f'TABLES={user_and_table_str}',
                'db_password': f'"{pdb_name}"',
                'instance_name': self.assist_ins_sid,
                'is_pdb': True,
                'type': 'expdp'
            }
            return self._execute_expdp_impdp_windows(expdp_params)
        else:
            oracle_user_env = get_oracle_user_env_by_linux(assist_ins_sid)
            cmd = (f"{oracle_user_env}"
                   f" expdp {auth_scope} directory={dir_name} dumpfile={dumpfile_name}"
                   f" tables={user_and_table_str}")
            return checkout_user_and_execute_cmd(self._pid, self.ora_db_user, cmd)

    def _dump_pdb_tables_19c(self, assist_ins_sid, pdb):
        """
        dump pdb数据库的表

        @param assist_ins_sid: 实例id
        @return: ScriptExitCode状态
        """
        logger.info(f"Execute _dump_pdb_tables_19c")
        db_user = self.db_user
        db_password = f'\\\"{self.db_user_pwd}\\\"'
        pdb_name = pdb.get('pdb_name')
        dir_name = pdb.get('dir_name')

        # 合并表结构 user_and_table_str ="u1.t1, u2,t2"
        user_and_table_str_list = []
        dumpfile_name = f"{pdb_name}_{self._pid}.dmp"
        for json_table in self.tables:
            if json_table.get('pdb_name') == pdb_name:
                table_name = escape_special_chars(json_table.get('table_name'))
                user_and_table = f"{json_table.get('user_name')}.{json_table.get('table_name')}"
                self.table_dump_file[user_and_table] = dumpfile_name
                if platform.system().lower() == "windows":
                    user_and_table_str_list.append(f"{json_table.get('user_name')}.\"{json_table.get('table_name')}\"")
                else:
                    user_and_table_str_list.append(f"{json_table.get('user_name')}.\\\"{table_name}\\\"")
        user_and_table_str = ",".join(user_and_table_str_list)

        if db_user and db_password:
            auth_scope = f"\\\'{db_user}/{db_password} as sysdba\\\'"
        else:
            auth_scope = f"userid=\\\"/ as sysdba\\\""
        userid = self.create_pdb_userid_params(pdb_name)
        if platform.system().lower() == "windows":
            expdp_params = {
                'auth_scope': f'USERID={userid}',
                'directory_scope': f'DIRECTORY={dir_name}',
                'dumpfile_scope': f'DUMPFILE={dumpfile_name}',
                'tables_scope': f'TABLES={user_and_table_str}',
                'db_password': f'"{self.db_user_pwd}"',
                'instance_name': self.assist_ins_sid,
                'is_pdb': True,
                'type': 'expdp'
            }
            return self._execute_expdp_impdp_windows(expdp_params)
        else:
            oracle_user_env = get_oracle_user_env_by_linux(assist_ins_sid, pdb_name)
            cmd = (f"{oracle_user_env}"
                   f" expdp {auth_scope} directory={dir_name} dumpfile={dumpfile_name}"
                   f" tables={user_and_table_str}")
            return checkout_user_and_execute_cmd(self._pid, self.ora_db_user, cmd)

    def create_pdb_userid_params(self, pdb_name):
        if self.target_sub_type == OracleDataBaseType.ORACLE_CLUSTER:
            return f"{self.assist_ins_sid.upper()}/{pdb_name}@{self.service_ip}:{self.port}/{pdb_name}"
        else:
            return f"{self.assist_ins_sid.upper()}/{pdb_name}@{pdb_name}"

    def _create_expdp_impdp_params_file(self, params):
        expdp_impdp_param_list = [
            'auth_scope',
            'directory_scope',
            'dumpfile_scope',
            'tables_scope',
            'remap_tablespace_scope',
            'remap_schema_scope',
            'exclude_scope',
            'table_exists_action_scope',
            'remap_table_scope'
        ]
        lines = [params.get(key, '') for key in expdp_impdp_param_list]
        par_file_content = '\n'.join(line for line in lines if line)
        params_file_path = os.path.join(PluginPathConstant.WINDOWS_TMP_PATH,
                                        f"{self.assist_ins_sid}_{params.get('type', '')}.par")
        with open(params_file_path, mode='w', encoding='utf-8') as par_file:
            par_file.write(par_file_content)
        return params_file_path

    def _execute_expdp_impdp_windows(self, params):
        params_file_path = self._create_expdp_impdp_params_file(params)
        content = f"set NLS_LANG=American_America.US7ASCII\n set ORACLE_SID={params.get('instance_name', '')} \n" \
                  f"{params.get('type', '')} parfile={params_file_path}"
        file_name = f"cmd_{self._pid}.bat"
        write_tmp_file(content, file_name)
        child = popen_spawn.PopenSpawn(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}", timeout=60)
        if params.get('is_pdb', False):
            index = child.expect(PexpectResult.EXPDP_IMPDP_SUCCEEDED, timeout=None)
            check_del_dir_or_file(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}")
            if index in (0, 1, 2):
                try:
                    error_output = child.before.decode()
                    logger.error(f"error_output:{error_output}")
                except Exception as exception:
                    logger.error(f"exception is: {exception}.")
                logger.error(f"execute {params.get('type', '')}_pdb  failed.")
                return ScriptExitCode.ERROR_PARAM_INVALID
        else:
            index = child.expect(PexpectResult.PASS_WORD)
            check_del_dir_or_file(f"{PluginPathConstant.WINDOWS_TMP_PATH}/{file_name}")
            logger.info("Delete login database bat success")
            if index in (0, 1):
                logger.error(f"execute {params.get('type', '')}  failed.")
                return ScriptExitCode.ERROR_PARAM_INVALID
            child.sendline(params.get('db_password', ''))
            index = child.expect(PexpectResult.EXPDP_IMPDP_SUCCEEDED, timeout=None)
            if index in (0, 1, 2):
                logger.error(f"execute {params.get('type', '')}  failed.")
                return ScriptExitCode.ERROR_PARAM_INVALID
        return ScriptExitCode.SUCCESS

    def _imp_table_to_pdb_19c(self, table):
        """
        单个表导入表到pdb数据库
        :param table:待导入的表信息
        :return:
        """
        table_name = table.get('table_name', '')
        table_space = table.get('table_space', '')
        user_name = table.get('user_name', '')
        table_full_name = f"{user_name}.'\\\"{table_name}\\\"'"
        target_table_name = f"'\\\"{table.get('target_table_name', '')}\\\"'"
        target_user_name = table.get('target_user_name', '')
        target_table_space = table.get('target_table_space', '')
        target_pdb_name = table.get('target_pdb_name', '')
        dump_file = self.table_dump_file.get(f"{user_name}.{table_name}")
        table_name = f"'\\\"{table_name}\\\"'"
        imp_dir = f"{self.assist_ins_sid}_expdp_dir"
        auth_scope = self._get_target_auth_scope()
        oracle_user_env = get_oracle_user_env_by_linux(self.db_instance, target_pdb_name)
        exclude_ref_constraint = " EXCLUDE=REF_CONSTRAINT" if table.get('is_has_foreign_key', '') else ''
        cmd = (f"{oracle_user_env}"
               f" impdp {auth_scope} directory={imp_dir} dumpfile={dump_file}"
               f" remap_tablespace={table_space}:{target_table_space} remap_schema={user_name}:{target_user_name}"
               f" remap_table={table_name}:{target_table_name}"
               f" tables={table_full_name}{exclude_ref_constraint}")
        return checkout_user_and_execute_cmd(self._pid, self.ora_db_user, cmd)

    def _imp_table_to_pdb(self, table):
        """
        单个表导入表到pdb数据库
        :param table:待导入的表信息
        :return:
        """
        table_name = table.get('table_name', '')
        table_space = table.get('table_space', '')
        user_name = table.get('user_name', '')
        dump_file = self.table_dump_file.get(f"{user_name}.{table_name}")
        target_table_name = f"'\\\"{table.get('target_table_name', '')}\\\"'"
        target_user_name = table.get('target_user_name', '')
        target_table_space = table.get('target_table_space', '')
        target_pdb_name = table.get('target_pdb_name', '')
        table_full_name = f"{user_name}.'\\\"{table_name}\\\"'"
        table_name = f"'\\\"{table_name}\\\"'"
        imp_dir = f"{self.assist_ins_sid}_expdp_dir"

        auth_scope = f"{self.assist_ins_sid}/{target_pdb_name}@{target_pdb_name}"

        oracle_user_env = get_oracle_user_env_by_linux(self.db_instance)
        exclude_ref_constraint = " EXCLUDE=REF_CONSTRAINT" if table.get('is_has_foreign_key', '') else ''
        cmd = (f"{oracle_user_env}"
               f" impdp {auth_scope} directory={imp_dir} dumpfile={dump_file}"
               f" remap_tablespace={table_space}:{target_table_space} remap_schema={user_name}:{target_user_name}"
               f" remap_table={table_name}:{target_table_name}"
               f" tables={table_full_name}{exclude_ref_constraint}")
        return checkout_user_and_execute_cmd(self._pid, self.ora_db_user, cmd)

    def _create_directories(self, db_login_params, ins_sid, dir_name, dir_path):
        """
        创建directory用于后续导入导出
        执行创建导出directory的sql命令，参考如下
        create directory {directory名称} as '{directory路径}';
        grant read,write on directory {directory名称} to public;

        @param ins_sid: 实例id
        @param dir_name: directory名称
        @param dir_path: directory路径
        @param db_login_params: 登录数据库参数
        @return: ScriptExitCode状态
        """
        logger.info(f"start create dir, sid: {ins_sid}, dir_name: {dir_name}, dir_path: {dir_path}.")

        result, child = login_oracle_database_restore(self._pid, ins_sid, db_login_params, self.ora_db_user)
        if not result:
            logger.error(f'exec create dir sql login error, res: {result}.')
            return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        try:
            # 创建数据泵目录
            sql = f"create directory {dir_name} as '{dir_path}';"
            return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.DIRECTORY_CREATED)
            if not return_code:
                logger.error(f'exec create dir sql error, return_code: {return_code}. sql: {sql}, stdout:{std_out}')
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED

            # 给数据泵目录权限
            sql = f"grant read,write on directory {dir_name} to public;"
            return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.GRANT_SUCCEEDED)
            if not return_code:
                logger.error(f'exec grant read,write on directory to public error, '
                             f'return_code: {return_code}. sql: {sql}, stdout:{std_out}')
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()
        return ScriptExitCode.SUCCESS

    def _delete_directories(self, db_login_params, ins_sid, dir_name, pdb_name_list=None):
        """
        删除指定实例的directory

        @param ins_sid: 实例id
        @param dir_name: directory名称
        @param pdb_name_list: pdb列表，不为None时会进入每个pdb去删除directory
        @param db_login_params: 数据库登录参数
        @return: ScriptExitCode状态
        """
        result, child = login_oracle_database_restore(self._pid, ins_sid, db_login_params, self.ora_db_user)
        if not result:
            logger.error(f'exec create dir sql login error, res: {result}.')
            return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        try:
            if not pdb_name_list:
                return self._delete_single_directory(child, dir_name)
            else:
                return self._delete_pdb_directories(child, pdb_name_list, dir_name)
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()
        return ScriptExitCode.SUCCESS

    def _delete_pdb_directories(self, child, pdb_name_list, dir_name):
        for pdb_name in pdb_name_list:
            # 切换pdb
            sql = f"alter session set container={pdb_name};"
            return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.SESSION_ALTERED)
            if not return_code:
                logger.error(f'exec alter session set container error, return_code: {return_code}. sql: {sql}')
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
            sql = f"drop directory {dir_name};"
            return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.DIRECTORY_DROPPED)
            if not return_code:
                logger.error(f'drop directory error, std_out: {std_out}. sql: {sql}')
            if not self.is_version_later_than_19c() or platform.system().lower() == "windows":
                sql = f"drop user {self.assist_ins_sid};"
                return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.USER_DROPPED)
                if not return_code:
                    logger.error(f'drop user error, std_out: {std_out}. sql: {sql}')
        return ScriptExitCode.SUCCESS

    def _delete_single_directory(self, child, dir_name):
        """
        执行删除导出directory的sql命令，参考如下
        drop directory {directory名称};

        @param child:
        @param dir_name: directory名称
        @return: ScriptExitCode状态
        """
        # 删除数据泵目录
        sql = f"drop directory {dir_name};"
        return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.DIRECTORY_DROPPED)
        if not return_code:
            logger.error(f'exec drop dir sql error pid {self._pid}, return_code: {return_code}. sql: {sql}')
            return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        return ScriptExitCode.SUCCESS

    def _import_non_cdb_tables(self, dir_name, table):
        """
        导入非cdb数据库的表, 支持以下场景：
            表重命名 REMAP_TABLE=hr.t1:t2
            更换表空间 REMAP_TABLESPACE=TBSHR:TBSSCOTT
            更换用户 remap_schema=hr:SCOTT

        @param dir_name: directory对象名称
        @param table: {
            "user_name": "原表用户名",
            "table_name": "原表名",
            "pdb_name": "原PDB名(可能为空)",
            "table_space": "原表空间",
            "target_pdb_name": "用户指定的PDB(可能为空)",
            "target_user_name": "用户指定的用户名",
            "target_table_space": "用户指定的表空间",
            "target_table_name": "用户指定的表名"
        }
        @return: ScriptExitCode状态
        """
        table_name = table.get('table_name', '')
        table_space = table.get('table_space', '')
        user_name = table.get('user_name', '')
        target_table_name = table.get('target_table_name', '')
        target_user_name = table.get('target_user_name', '')
        target_table_space = table.get('target_table_space', '')
        dump_file = self.table_dump_file.get(f"{user_name}.{table_name}")
        table_full_name = f"{user_name}.'\\\"{table_name}\\\"'"
        auth_scope = self._get_target_auth_scope()

        if platform.system().lower() == "windows":
            db_user, db_user_pwd = self._get_target_db_user_pwd()
            impdp_params = {
                'auth_scope': f'USERID="{db_user}/ as sysdba"'
                if db_user and db_user_pwd
                else f'USERID="/ as sysdba"',
                'directory_scope': f'DIRECTORY={dir_name}',
                'dumpfile_scope': f'DUMPFILE={dump_file}',
                'remap_tablespace_scope': f'REMAP_TABLESPACE={table_space}:{target_table_space}',
                'remap_schema_scope': f'REMAP_SCHEMA={user_name}:{target_user_name}',
                'remap_table_scope': f'REMAP_TABLE=\'"{table_name}":"{target_table_name}"\'',
                'tables_scope': f'TABLES={user_name}."{table_name}"',
                'exclude_scope': "EXCLUDE=REF_CONSTRAINT" if table.get('is_has_foreign_key', '') else '',
                'table_exists_action_scope': "TABLE_EXISTS_ACTION=REPLACE" if self.is_overwrite else '',
                'instance_name': self.db_instance,
                'db_password': f'"{db_user_pwd}"',
                'type': 'impdp'
            }
            return self._execute_expdp_impdp_windows(impdp_params)
        else:
            table_name = f"'\\\"{table_name}\\\"'"
            target_table_name = f"'\\\"{table.get('target_table_name', '')}\\\"'"
            oracle_user_env = get_oracle_user_env_by_linux(self.db_instance)
            exclude_ref_constraint = " EXCLUDE=REF_CONSTRAINT" if table.get('is_has_foreign_key', '') else ''
            cmd = (f"{oracle_user_env}"
                   f" impdp {auth_scope} directory={dir_name} dumpfile={dump_file}"
                   f" remap_tablespace={table_space}:{target_table_space} remap_schema={user_name}:{target_user_name}"
                   f" remap_table={table_name}:{target_table_name}"
                   f" tables={table_full_name}{exclude_ref_constraint}")
            return checkout_user_and_execute_cmd(self._pid, self.ora_db_user, cmd)

    def _get_target_auth_scope(self):
        db_user = get_env_variable(f'job_targetObject_auth_authKey_{self._pid}')
        db_user_pwd = get_env_variable(f'job_targetObject_auth_authPwd_{self._pid}')
        db_password = f'\\\"{db_user_pwd}\\\"'
        if db_user and db_user_pwd:
            return f"\\\'{db_user}/{db_password} as sysdba\\\'"
        return f"userid=\\\"/ as sysdba\\\""

    def _get_target_db_user_pwd(self):
        target_db_user = get_env_variable(f'job_targetObject_auth_authKey_{self._pid}')
        target_db_password = get_env_variable(f'job_targetObject_auth_authPwd_{self._pid}')
        return target_db_user, target_db_password

    def _export_cdb_tables(self, expdp_dir_name, expdp_dir_path, pdb_list):
        for pdb_dir in pdb_list:
            return_code = self._exec_dump_pdb_tables(self.assist_ins_sid, pdb_dir)
            if return_code != ScriptExitCode.SUCCESS:
                logger.error(f"dump_pdb_tables failed: {pdb_dir.get('pdb_name')}")
                return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        return ScriptExitCode.SUCCESS

    def _export_non_cdb_tables(self, assist_ins_sid, assist_ins_expdp_dir_name):
        """
        dump非cdb数据库的表

        @param assist_ins_sid: 实例id
        @param assist_ins_expdp_dir_name: directory名称
        @return: ScriptExitCode状态
        """
        logger.info(f"start export non cdb tables, sid: {self._pid}.")
        db_user = self.db_user
        db_password = f'\\\"{self.db_user_pwd}\\\"'

        if db_user and db_password:
            auth_scope = f"\\\'{db_user}/{db_password} as sysdba\\\'"
        else:
            auth_scope = f"userid=\\\"/ as sysdba\\\""

        # 合并表结构 user_and_table_str ="u1.t1, u2,t2"
        dumpfile_name = "non_cdb_table.dmp"
        user_and_table_str_list = []
        for json_table in self.tables:
            if platform.system().lower() == "windows":
                user_and_table_str_list.append(f"{json_table.get('user_name')}.\"{json_table.get('table_name')}\"")
            else:
                table_name = escape_special_chars(json_table.get('table_name'))
                user_and_table_str_list.append(f"{json_table.get('user_name')}.\\\"{table_name}\\\"")
            user_and_table = f"{json_table.get('user_name')}.{json_table.get('table_name')}"
            self.table_dump_file[user_and_table] = dumpfile_name
        user_and_table_str = ",".join(user_and_table_str_list)
        if platform.system().lower() == "windows":
            db_password = f'"{self.db_user_pwd}"'
            expdp_params = {
                'auth_scope': f'USERID="{db_user}/ as sysdba"'
                if db_user and db_password
                else f'USERID="/ as sysdba"',
                'directory_scope': f'DIRECTORY={assist_ins_expdp_dir_name}',
                'dumpfile_scope': f'DUMPFILE={dumpfile_name}',
                'tables_scope': f'TABLES={user_and_table_str}',
                'instance_name': assist_ins_sid,
                'db_password': f'{db_password}',
                'type': 'expdp'
            }
            return self._execute_expdp_impdp_windows(expdp_params)
        else:
            oracle_user_env = get_oracle_user_env_by_linux(assist_ins_sid)
            cmd = (f"{oracle_user_env}"
                   f" expdp {auth_scope} directory={assist_ins_expdp_dir_name} dumpfile={dumpfile_name}"
                   f" tables={user_and_table_str}")
            return checkout_user_and_execute_cmd(self._pid, self.ora_db_user, cmd)

    def _is_cdb_database_copy(self):
        """
        检查是否为cdb副本恢复

        @return: 是否为cdb数据库恢复
        """
        json_tables = self.tables
        first_pdb_name = json_tables[0].get("pdb_name", None)
        if first_pdb_name and first_pdb_name != "":
            return True
        else:
            return False

    def _is_cdb_database_restore_aim(self):
        if self.is_cdb_aim is None:
            self.is_cdb_aim = check_cdb_is_open(self._pid, self.login_params)
        return self.is_cdb_aim

    def _open_all_pdbs(self):
        logger.info(f"open_all_pdbs")
        cmd = f"alter pluggable database all open;"
        db_login_params = {
            'pid': self._pid,
            "instance_name": self.assist_ins_sid,
            "db_install_user": self.ora_db_user,
            'auth_pwd': f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_copies_0_protectObject_auth_authKey_{self._pid}"
        }
        logger.info(f"_open_all_pdbs db_login_params {db_login_params}")
        ret = exec_oracle_sql_cmd(db_login_params, cmd, PexpectResult.PDB_OPENED)
        if ret:
            logger.info(f"_open_all_pdbs success.")
            return True
        restricted_pdb_name_list = self.get_restricted_pdb_name_list(self.assist_ins_sid, self.ora_db_user)
        if not restricted_pdb_name_list:
            logger.info(f"get restricted_pdb_name_list null.")
            return False
        table_space_name = self.get_table_space_name()
        if not table_space_name:
            logger.error(f"get table_space_name failed.")
            return False
        for pdb in restricted_pdb_name_list:
            ret = self.create_table_space(pdb, table_space_name, self.assist_ins_sid, self.ora_db_user)
            if not ret:
                logger.error(f"cannot create_table_space {table_space_name} in pdb {pdb}.")
                return False
            ret = self.close_pdb(pdb, self.assist_ins_sid, self.ora_db_user)
            if not ret:
                logger.error(f"cannot close_pdb {pdb}.")
                return False
            ret = self.open_pdb(pdb, self.assist_ins_sid, self.ora_db_user)
            if not ret:
                logger.error(f"cannot open_pdb {pdb}.")
                return False
        return True

    def _delete_exists_table(self, table):
        if self.is_overwrite and self._table_exists(table):
            # 覆盖场景需要删除原表
            ret = self._drop_table(table)
            if ret != ScriptExitCode.SUCCESS:
                logger.error(f'_delete_exists_table failed')
                return ret
        return ScriptExitCode.SUCCESS

    def _import_cdb_tables(self):
        logger.info(f'_import_cdb_tables')
        json_tables = self.tables
        for json_table in json_tables:
            ret = self._delete_exists_table(json_table)
            if ret != ScriptExitCode.SUCCESS:
                logger.error(f'_delete_exists_table failed')
                return ret
            return_code = self._exec_imp_pdb_table(json_table)
            if return_code != ScriptExitCode.SUCCESS:
                logger.error(f'_imp_table_to_pdb failed')
                return return_code
        return ScriptExitCode.SUCCESS

    def _import_cdb_tables_windows(self):
        logger.info(f'_import_cdb_tables_windows')
        ret = ScriptExitCode.SUCCESS
        json_tables = self.tables
        _, db_user_pwd = self._get_target_db_user_pwd()
        user_and_table_str_list = []
        remap_table_str_list = []
        grouped_tables = {}
        for table in json_tables:
            key = (table.get('pdb_name'), table.get('target_pdb_name'), table.get('table_space'),
                   table.get('target_table_space'), table.get('user_name'),
                   table.get('target_user_name'), table.get('is_has_foreign_key'))
            if key not in grouped_tables:
                grouped_tables[key] = []
            grouped_tables[key].append(table)
        for group_key, tables in grouped_tables.items():
            (original_pdb_name, pdb_name, table_space, target_table_space, user_name, target_user_name,
             is_has_foreign_key) = group_key
            for table in tables:
                user_and_table_str_list.append(f"{table.get('user_name')}.\"{table.get('table_name')}\"")
                remap_table_str_list.append(f"'\"{table.get('table_name')}\":\"{table.get('target_table_name')}\"'")
            user_and_table_str = ",".join(user_and_table_str_list)
            remap_table_str = ",".join(remap_table_str_list)
            params = {
                'originalPdbName': f'{original_pdb_name}',
                'pdbName': f'{pdb_name}',
                'tableSpace': f'{table_space}',
                'targetTableSpace': f'{target_table_space}',
                'userName': f'{user_name}',
                'targetUserName': f'{target_user_name}',
                'remapTableStr': f'{remap_table_str}',
                'userAndTableStr': f'{user_and_table_str}',
                'isHasForeignKey': f'{is_has_foreign_key}',
                'dbUserPwd': f'{db_user_pwd}'
            }
            exec_cnt = 0
            ret = ScriptExitCode.SUCCESS
            while exec_cnt < 20:
                ret = self._exec_imp_pdb_table_windows(params)
                if ret == ScriptExitCode.SUCCESS:
                    logger.info(f"exec_imp_pdb_table SUCCESS {exec_cnt}")
                    break
                exec_cnt += 1
                time.sleep(10)
            user_and_table_str_list.clear()
            remap_table_str_list.clear()
            if ret != ScriptExitCode.SUCCESS:
                break
        return ret

    def _exec_imp_pdb_table_windows(self, params):
        original_pdb_name = params.get('originalPdbName', '')
        pdb_name = params.get('pdbName', '')
        table_space = params.get('tableSpace', '')
        target_table_space = params.get('targetTableSpace', '')
        user_name = params.get('userName', '')
        target_user_name = params.get('targetUserName', '')
        remap_table_str = params.get('remapTableStr', '')
        user_and_table_str = params.get('userAndTableStr', '')
        is_has_foreign_key = params.get('isHasForeignKey', '')
        db_user_pwd = params.get('dbUserPwd', '')
        userid = self.create_pdb_userid_params(pdb_name)
        impdp_params = {
            'auth_scope': f'USERID={userid}',
            'directory_scope': f'DIRECTORY={self.assist_ins_sid}_expdp_dir',
            'dumpfile_scope': f'DUMPFILE={original_pdb_name}_{self._pid}.dmp',
            'remap_tablespace_scope': f'REMAP_TABLESPACE={table_space}:{target_table_space}',
            'remap_schema_scope': f'REMAP_SCHEMA={user_name}:{target_user_name}',
            'remap_table_scope': f'REMAP_TABLE={remap_table_str}',
            'tables_scope': f'TABLES={user_and_table_str}',
            'exclude_scope': "EXCLUDE=REF_CONSTRAINT" if is_has_foreign_key else '',
            'table_exists_action_scope': "TABLE_EXISTS_ACTION=REPLACE" if self.is_overwrite else '',
            'instance_name': self.db_instance,
            'db_password': f'"{db_user_pwd}"',
            'is_pdb': True,
            'type': 'impdp'
        }
        ret = self._execute_expdp_impdp_windows(impdp_params)
        return ret

    def _exec_import_tables(self, expdp_dir_name, expdp_dir_path):
        target_pdb_names = set()
        if self._is_cdb_database_restore_aim():
            logger.info(f'cdb exec_import_tables')
            target_pdb_names = self._get_target_pdb_names()
            if len(target_pdb_names) == 0:
                logger.error(f'target_pdb_names empty')
                return ScriptExitCode.ERROR_PARAM_INVALID, target_pdb_names
            ret = self._create_target_pdb_imp_dirs(target_pdb_names, expdp_dir_name)
            if ret != ScriptExitCode.SUCCESS:
                logger.error(f'_create_target_pdb_dirs_and_users failed')
                return ret, target_pdb_names
            if platform.system().lower() == "windows":
                ret = self._import_cdb_tables_windows()
                if ret != ScriptExitCode.SUCCESS:
                    logger.error(f'_import_cdb_tables_windows failed')
                    return ret, target_pdb_names
            else:
                ret = self._import_cdb_tables()
                if ret != ScriptExitCode.SUCCESS:
                    logger.error(f'_import_cdb_tables failed')
                    return ret, target_pdb_names
        else:
            # 目标为非cdb场景
            # 在目标实例执行sql创建导出目录
            logger.info(f'no cdb exec_import_tables, instance:{self.db_instance}')
            db_login_params = {
                'auth_pwd': f"job_targetObject_auth_authPwd_{self._pid}",
                'auth_key': f"job_targetObject_auth_authKey_{self._pid}"
            }
            ret = self._create_directories(db_login_params, self.db_instance, expdp_dir_name, expdp_dir_path)
            if ret != ScriptExitCode.SUCCESS:
                return ret, set()
            # 导入表
            json_tables = self.tables
            for json_table in json_tables:
                ret = self._delete_exists_table(json_table)
                if ret != ScriptExitCode.SUCCESS:
                    logger.error(f'_delete_exists_table failed')
                    return ret, set()
                ret = self._import_non_cdb_tables(expdp_dir_name, json_table)
                if ret != ScriptExitCode.SUCCESS:
                    logger.error(f'_import_non_cdb_tables failed')
                    return ret, set()
        return ScriptExitCode.SUCCESS, target_pdb_names

    def _create_pdb_directories(self, db_login_params, ins_sid, dir_name, dir_path, pdb_name_list):
        """
        pdb数据库下创建directory用于后续导入导出

        @param ins_sid: 实例id
        @param dir_name: directory名称
        @param dir_path: directory路径
        @param pdb_name_list: pdb列表，不为None时会进入每个pdb去创建directory
        @param db_login_params: 数据库登录参数
        @return: pdb_list
        """
        logger.info(f'_create_pdb_directories, pdb_name_list: {pdb_name_list}')
        result, child = login_oracle_database_restore(self._pid, ins_sid, db_login_params, self.ora_db_user)
        if not result:
            logger.error(f'exec create dir sql login error, res: {result}.')
            return []
        pdb_list = []
        try:
            for pdb_name in pdb_name_list:
                pdb_dir = self._create_pdb_directory(child, dir_name, dir_path, pdb_name)
                if not pdb_dir:
                    logger.error(f'_create_pdb_directories failed.')
                    return []
                pdb_list.append(pdb_dir)
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()
        return pdb_list

    def _create_pdb_directory(self, child, dir_name, dir_path, pdb_name):
        logger.info(f'_create_pdb_directory')
        user_name = self.assist_ins_sid
        sql = f"alter session set container={pdb_name};"
        return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.SESSION_ALTERED)
        if not return_code:
            logger.error(f'exec alter session set container error, std_out: {std_out}. sql: {sql}')
            return {}
        if not self.is_version_later_than_19c() or platform.system().lower() == "windows":
            sql = f"grant dba to {user_name} IDENTIFIED BY {pdb_name};"
            return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.GRANT_SUCCEEDED)
            if not return_code:
                logger.error(f'exec grant user error, std_out: {std_out}. sql: {sql}')
                return {}
        sql = f"create directory {dir_name} as '{dir_path}';"
        return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.DIRECTORY_CREATED)
        if not return_code:
            logger.error(f'create directory error, std_out: {std_out}. sql: {sql}')
            return {}
        if not self.is_version_later_than_19c() or platform.system().lower() == "windows":
            sql = f"grant read,write on directory {dir_name} to {user_name};"
        else:
            sql = f"grant read,write on directory {dir_name} to public;"

        return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.GRANT_SUCCEEDED)
        if not return_code:
            logger.error(f'grant directory error, std_out: {std_out}. sql: {sql}')
            return {}
        pdb = {
            'pdb_name': pdb_name,
            'dir_name': dir_name,
            'dir_path': dir_path,
            'pdb_user': user_name,
        }
        return pdb

    def _create_target_pdb_imp_dirs(self, target_pdb_names, impdp_dir_name):
        logger.info(f'_create_target_pdb_imp_dirs')
        # 在恢复目标pdb下创建dir用于导入
        if platform.system().lower() == "windows":
            impdp_dir_path = os.path.join(self.oracle_home, f"table_dump_{self._job_id}")
        else:
            impdp_dir_path = f"{self.cache_repo_path}/table_dump_{self._job_id}"
        db_login_params = {
            'auth_pwd': f"job_targetObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_targetObject_auth_authKey_{self._pid}"
        }
        if not self._create_pdb_directories(
                db_login_params, self.db_instance, impdp_dir_name, impdp_dir_path, target_pdb_names):
            logger.error(f'_create_pdb_directories_and_users failed')
            return ScriptExitCode.ERROR_ORACLE_EXESQL_FAILED
        return ScriptExitCode.SUCCESS

    def get_restricted_pdb_name_list(self, db_instance_name, install_user_name):
        logger.info(f"get_restricted_pdb_name_list, database instance name: {db_instance_name}, "
                    f"install user name: {install_user_name}, pid: {self._pid}.")
        cmd = "show pdbs;"
        result, child = login_oracle_database(self._pid, db_instance_name, install_user_name, timeout=30)
        if not result:
            return set()
        try:
            return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
            if return_code:
                pdb_list = parse_html_result(std_out)
                pdb_name_set = get_pdb_name_set(pdb_list)
                return pdb_name_set
            logger.error(f"Failed to execute cmd: {cmd}, pid: {self._pid}.")
            return set()
        except Exception as exception:
            logger.error(f"Failed to execute cmd: {cmd}, pid: {self._pid}, exception is: {exception}.")
            return set()
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()

    def create_table_space(self, pdb_name, table_space_name_list, db_instance_name, install_user_name):
        logger.info(f"get_all_table_space_from_pdb, database instance name: {db_instance_name}, "
                    f"install user name: {install_user_name}, pid: {self._pid}.")
        result, child = login_oracle_database(self._pid, db_instance_name, install_user_name, timeout=30)
        cmd = f"alter session set container='{pdb_name}';\n"
        for table_space_name in table_space_name_list:
            datafile_path = os.path.join(self.oracle_home, f"temp_{table_space_name}_{db_instance_name}.dbf")
            cmd += f"CREATE TABLESPACE {table_space_name} DATAFILE " \
                   f"'{datafile_path}' SIZE 100M AUTOEXTEND ON;\n"
        try:
            return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.CREATE_TABLESPACE_RES)
            if return_code:
                return True
            logger.error(f"Failed to execute cmd: {cmd}, pid: {self._pid}.")
            return False
        except Exception as exception:
            logger.error(f"Failed to execute cmd: {cmd}, pid: {self._pid}, exception is: {exception}.")
            return False
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()

    def get_message_set(self, db_instance_name, install_user_name):
        logger.info(f"get_all_table_space_from_pdb, database instance name: {db_instance_name}, "
                    f"install user name: {install_user_name}, pid: {self._pid}.")
        cmd = f"set linesize 999;\n select CAUSE, CON_ID, MESSAGE from pdb_plug_in_violations;"
        result, child = login_oracle_database(self._pid, db_instance_name, install_user_name, timeout=30)
        if not result:
            return set()
        try:
            return_code, std_out = exec_sql_cmd(child, cmd, PexpectResult.FILE_PATH)
            if return_code:
                violations = parse_html_result(std_out)
                message_set = get_violation_message_set(violations)
                return message_set
            logger.error(f"Failed to execute cmd: {cmd}, pid: {self._pid}.")
            return set()
        except Exception as exception:
            logger.error(f"Failed to execute cmd: {cmd}, pid: {self._pid}, exception is: {exception}.")
            return set()
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()

    def get_table_space_name(self):
        logger.info(f"start get_table_space_name.")
        message_set = self.get_message_set(self.assist_ins_sid, self.ora_db_user)
        table_space_name_list = set()
        if not message_set:
            return table_space_name_list
        for message in message_set:
            table_space_name_list.add(str(message).split("quota unlimited on")[1].strip()[:-1])
        logger.info(f"get_table_space_name table_space_name_list {table_space_name_list}.")
        return table_space_name_list

    def close_pdb(self, pdb_name, db_instance_name, install_user_name):
        logger.info(f"close_pdb, database instance name: {db_instance_name}, "
                    f"install user name: {install_user_name}, pid: {self._pid}.")
        result, child = login_oracle_database(self._pid, db_instance_name, install_user_name, timeout=30)
        sql = f"alter pluggable database {pdb_name} close immediate;"
        return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.CLOSE_PDB)
        if not return_code:
            logger.error(f'alter pluggable error, return_code: {return_code}. sql: {sql}, stdout:{std_out}')
            return False
        return True

    def open_pdb(self, pdb_name, db_instance_name, install_user_name):
        logger.info(f"open_pdb, database instance name: {db_instance_name}, "
                    f"install user name: {install_user_name}, pid: {self._pid}.")
        result, child = login_oracle_database(self._pid, db_instance_name, install_user_name, timeout=30)
        sql = f"alter pluggable database {pdb_name} open;"
        return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.PDB_OPENED)
        if not return_code:
            logger.error(f'alter pluggable error, return_code: {return_code}. sql: {sql}, stdout:{std_out}')
            return False
        return True
